package com.ice.demo.reactive;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @author mayf
 * @version 1.0
 * @description
 * @create 2025-10-11 23:03
 */
public class ReactiveStreamDemo {

    public static void main(String[] args) {

        // 1.创建一个消息发布者
        SubmissionPublisher publisher = new SubmissionPublisher();
//        Flow.Publisher publisher = new Flow.Publisher() {
//            @Override
//            public void subscribe(Flow.Subscriber subscriber) {
//                System.out.println("订阅");
//                subscriber.onSubscribe();
//            }
//        };
        // 2.创建一个消息订阅者
        Flow.Subscriber subscriber = new Flow.Subscriber() {

            Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                System.out.println("建立订阅链接");
                this.subscription = subscription;
                this.subscription.request(1); //第一次需要
                System.out.println("-----");
            }

            @Override
            public void onNext(Object item) {
                System.out.println("接受数据 item = " + item);
                // 业务处理
                this.subscription.request(2); //背压
            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("发生错误了");
            }

            @Override
            public void onComplete() {
                System.out.println("数据接收完了");
            }
        };

        // 3.建立订阅关系
        publisher.subscribe(subscriber);

        // 4.发送数据
        publisher.submit("hello reactive stream1");
        publisher.submit("hello reactive stream2");
        publisher.submit("hello reactive stream3");
        publisher.submit("hello reactive stream4");



        try {
//            Thread.currentThread().join();
            Thread.sleep(500);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

    }
}
